#pragma once

#include "blue.hpp"
#include "devManager.hpp"
#include "mqttClient.hpp"
#include "msg.hpp"
#include "threadPool.hpp"

#include <atomic>
#include <memory>
#include <regex>

#include "spdlog/async.h"
#include "spdlog/common.h"
#include "spdlog/sinks/daily_file_sink.h"
#include "spdlog/sinks/stdout_color_sinks.h"
#include "spdlog/spdlog.h"

using MqttTopicRouteTable_t = std::vector<std::pair<std::regex, std::function<bool(const Msg_t &)>>>;

class MainProcess {
private:
    inline static const std::string mqttClientId = "modbus";
    inline static const std::string mqttHostUrl = "localhost";
    inline static constexpr std::uint16_t mqttPort = 1883;

    inline static const std::string devModbusSrc = "dev_rs485";

    inline static const std::string dev645Type = "DL/T645";
    inline static const std::string devModbusType = "Modbus_RTU";

    inline static const std::string devModbusRlistTopic = "dev/dev_rs485/sys_agent/rlist";
    inline static const std::string devRlistAckTopic = "dev/sys_agent/dev_rs485/rlistack";
    inline static const std::string devReadTopic = "dev/+/+/read";
    inline static const std::string devWriteTopic = "dev/+/+/write";
    inline static const std::string devAddTopic = "dev/sys_agent/dev_rs485/add";
    inline static const std::string devDeleteTopic = "dev/sys_agent/dev_rs485/delete";
    inline static const std::string devAutoAddTopic = "dev/+/dev_rs485/autoadd";
    inline static const std::string devAutoAddAckTopic = "dev/dev_rs485/web/autoaddack";
    inline static const std::string devModbus0InfoTopic = "dev/dev_rs485/0/info";
    inline static const std::string devOtaTopic = "dev/+/+/ota";

    inline static constexpr std::size_t threadPoolSize = 4;

private:
    std::shared_ptr<MqttClient> _mqtt;
    std::shared_ptr<ThreadPool> _pool;
    std::shared_ptr<DevManager> _device;
    std::shared_ptr<BlueTooth> _blueTooth;

    std::atomic<bool> _isRecvModbusRlistAck {false};
    std::atomic<bool> _isRecvDl645RlistAck {false};

    std::mutex _releaseMutex;
    std::condition_variable _releaseCv;
    std::vector<std::size_t> _releaseDeviceList;

private:
    std::vector<std::string> split(const std::string &source, const std::string &s);

    MqttTopicRouteTable_t mqttRouteTable; /* 路由表 */

private:
    void onAddModbusDevice(std::shared_ptr<ModbusDevice_t> device);
    void onDelModbusDevice(std::shared_ptr<ModbusDevice_t> device);

private:
    void publish0Info(void *dev, const std::string &name, OldData_t &oldData, NewData_t &newData);
    void onLeScanDevice(const std::string &name, const std::string &mac);
    void onLineStatChange(void *arg);

    std::vector<uint8_t> mac2vec(const std::string &mac);

private:
    bool mqttMsgParse(const Msg_t &msg);
    bool parseRlistMsg(const Msg_t &msg);
    bool devReadById(const Msg_t &msg);
    bool devReadByModel(const Msg_t &msg);
    bool devWriteById(const Msg_t &msg);
    bool devDeleteById(const Msg_t &msg);
    bool devAddMsgParse(const Msg_t &msg);
    bool devAutoAddmsgParse(const Msg_t &msg);
    bool devOtaMsgParse(const Msg_t &msg);

public:
    void waitMqttMsgThread(void);
    void requestModbusRlistThread(void);
    void request645RlistThread(void);
    void releaseDeviceListThread(void);

public:
    MainProcess();
};

MainProcess::MainProcess()
{
    mqttRouteTable = {
        {std::regex("^dev/[A-Za-z_0-9]+/[A-Za-z_0-9]+/rlistack$"), std::bind(&MainProcess::parseRlistMsg, this, std::placeholders::_1)},
        {std::regex("^dev/[A-Za-z_0-9]+/[0-9]+/read$"), std::bind(&MainProcess::devReadById, this, std::placeholders::_1)},
        {std::regex("^dev/[A-Za-z_0-9]+/[^0-9]+[0-9]+/read$"), std::bind(&MainProcess::devReadByModel, this, std::placeholders::_1)},
        {std::regex("^dev/.+/.+/add$"), std::bind(&MainProcess::devAddMsgParse, this, std::placeholders::_1)},
        {std::regex("^dev/.+/[0-9]+/write$"), std::bind(&MainProcess::devWriteById, this, std::placeholders::_1)},
        {std::regex("^dev/.+/.+/delete$"), std::bind(&MainProcess::devDeleteById, this, std::placeholders::_1)},
        {std::regex("^dev/.+/.+/autoadd$"), std::bind(&MainProcess::devAutoAddmsgParse, this, std::placeholders::_1)},
        {std::regex("^dev/.+/.+/ota$"), std::bind(&MainProcess::devOtaMsgParse, this, std::placeholders::_1)},
    };

    main_info("self process version --> {}", MODBUS_VERSION);
    main_info("self log file --> {}", LOG_FILE_NAME);

    _mqtt = std::make_shared<MqttClient>(mqttClientId);
    _pool = std::make_shared<ThreadPool>(threadPoolSize);
    _device = std::make_shared<DevManager>();

    _blueTooth = std::make_shared<BlueTooth>();
    _blueTooth->registerOnLeScan(std::bind(&MainProcess::onLeScanDevice, this, std::placeholders::_1, std::placeholders::_2));

    _mqtt->connect(mqttHostUrl, mqttPort);

    _mqtt->registerOnConnected(std::bind(&MqttClient::subscribe, _mqtt.get(), devRlistAckTopic));
    _mqtt->registerOnConnected(std::bind(&MqttClient::subscribe, _mqtt.get(), devReadTopic));
    _mqtt->registerOnConnected(std::bind(&MqttClient::subscribe, _mqtt.get(), devAddTopic));
    _mqtt->registerOnConnected(std::bind(&MqttClient::subscribe, _mqtt.get(), devWriteTopic));
    _mqtt->registerOnConnected(std::bind(&MqttClient::subscribe, _mqtt.get(), devAutoAddTopic));
    _mqtt->registerOnConnected(std::bind(&MqttClient::subscribe, _mqtt.get(), devDeleteTopic));
    _mqtt->registerOnConnected(std::bind(&MqttClient::subscribe, _mqtt.get(), devOtaTopic));

    _device->registerOnAddModbusDevice(std::bind(&MainProcess::onAddModbusDevice, this, std::placeholders::_1));
    _device->registerOnDelModbusDevice(std::bind(&MainProcess::onDelModbusDevice, this, std::placeholders::_1));

    std::thread(&MainProcess::waitMqttMsgThread, this).detach();
    std::thread(&MainProcess::requestModbusRlistThread, this).detach();
    // std::thread(&MainProcess::request645RlistThread, this).detach();
    std::thread(&MainProcess::releaseDeviceListThread, this).detach();
}

bool MainProcess::mqttMsgParse(const Msg_t &msg)
{
    for (const auto &[re, callback] : mqttRouteTable) {
        if (std::regex_match(msg.topic, re)) {
            return callback(msg);
        }
    }

    main_error("not suppose topic --> {}", msg.topic);

    return true;
}

bool MainProcess::parseRlistMsg(const Msg_t &msg)
{
    auto ack = json::parse(msg.payload).get<Protocol::RlistAckMsg_t>();

    if (ack.dev.list.type == dev645Type) {
        _isRecvDl645RlistAck = true;
        /* TODO: */

    } else if (ack.dev.list.type == devModbusType) {
        _isRecvModbusRlistAck = true;

        for (auto &dev : ack.dev.devices) {
            _device->addModbusDevice(dev);
        }
    }

    return true;
}

bool MainProcess::devReadById(const Msg_t &msg)
{
    auto result = split(msg.topic, "/");
    auto &dest = result[2];
    auto &src = result[1];

    auto read = json::parse(msg.payload).get<Protocol::ReadIdMsg_t>();
    Protocol::ReadIdMsgAck_t ack = read;
    ack.setSrc(devModbusSrc);

    for (auto &dev : ack._dev.devices) {
        dev.setStatus(false);
        auto d = _device->findModbusDeviceById(dev.id);
        if (!d) {
            main_error("can't find device id --> {}", dev.id);
            return true;
        }

        for (auto &p : dev.points) {
            if (d->readPoint(p.key)) {
                p.setStatus(true);
                dev.setStatus(true);
            } else {
                p.setStatus(false);
            }

            auto point = d->pointTable.findPoint(p.key);

            if (point.value()->type == TYPE_STR) {
                p.setKeyValue(point.value()->rData.get<std::string>());
            } else if (point.value()->type == TYPE_STR_HEX) {
                p.setKeyValue(point.value()->rData.toHexString());
            } else if (point.value()->type == TYPE_I16 || point.value()->type == TYPE_I8 || point.value()->type == TYPE_I32) {
                if (point.value()->unit.has_value()) {
                    p.setKeyValue(point.value()->rData.get<int>() * point.value()->unit.value());
                } else {
                    p.setKeyValue(point.value()->rData.get<int>());
                }
            } else {
                if (point.value()->unit.has_value()) {
                    p.setKeyValue(point.value()->rData.get<uint32_t>() * point.value()->unit.value());
                } else {
                    p.setKeyValue(point.value()->rData.get<uint32_t>());
                }
            }
        }
    }

    return _mqtt->publish(result[0] + "/" + dest + "/" + src + "/rack", json(ack).dump());
}

bool MainProcess::devReadByModel(const Msg_t &msg)
{
    auto result = split(msg.topic, "/");
    auto &dest = result[2];
    auto &src = result[1];

    if (dest != devModbusSrc) {
        return true;
    }

    auto read = json::parse(msg.payload).get<Protocol::ReadModelMsg_t>();

    for (const auto &dev : read._dev.devices) {
        auto devList = _device->findModbusDeviceByModel(dev.model);

        if (devList.empty()) {
            continue;
        }

        for (const auto &point : dev.points) {
            for (const auto &d : devList) {
                d->readPoint(point.key);
            }
        }
    }

    return true;
}

bool MainProcess::devWriteById(const Msg_t &msg)
{
    auto result = split(msg.topic, "/");
    auto &dest = result[2];
    auto &src = result[1];

    auto write = json::parse(msg.payload).get<Protocol::WriteIdMsg_t>();
    Protocol::WriteIdMsgAck_t ack = write;
    ack.setSrc(devModbusType);

    for (auto &dev : ack._dev.devices) {
        auto d = _device->findModbusDeviceById(dev.id);
        dev.setStatus(false);
        if (!d) {
            main_error("can't find device id --> {}", dev.id);
            return true;
        }

        for (auto &p : dev.points) {
            if (d->writePoint(p.key, p.val)) {
                p.setStatus(true);
                dev.setStatus(true);
            } else {
                p.setStatus(false);
            }
        }
    }

    std::string writeIdAckTopic = "dev/" + dest + "/" + src + "/wack";

    return _mqtt->publish(writeIdAckTopic, json(ack).dump());
}

bool MainProcess::devAddMsgParse(const Msg_t &msg)
{
    auto result = split(msg.topic, "/");
    auto &dest = result[2];
    auto &src = result[1];

    if (dest != devModbusSrc) {
        return true;
    }

    auto add = json::parse(msg.payload).get<Protocol::AddMsg_t>();
    Protocol::AddMsgAck_t ack = add;
    ack.setSrc(devModbusSrc);

    for (auto &dev : ack._dev.devices) {
        if (dev.type == devModbusType) {
            ModbusDevice_t d = dev;
            d.setModbusRtu(_device->getRtu());
            auto point = d.pointTable.findFirstReadAblePoint();
            if (!point.has_value()) {
                dev.setStatus(false).setLog("can't find first readable point");
                continue;
            }

            auto result = d.readPoint(point.value()->first, true);
            if (result) {
                dev.setStatus(result);
                _device->addModbusDevice(d);
            } else {
                dev.setStatus(result).setLog("read point failed and point name --> " + point.value()->first);
            }
        } else if (dev.type == dev645Type) {
            /* TODO: add 645 device */
        }
    }

    std::string ackTopic = "dev/" + dest + "/" + src + "/addack";

    return _mqtt->publish(ackTopic, json(ack).dump());
}

bool MainProcess::devAutoAddmsgParse(const Msg_t &msg)
{
    auto result = split(msg.topic, "/");
    auto &dest = result[2];
    auto &src = result[1];

    if (dest != devModbusSrc) {
        return true;
    }

    auto autoAdd = json::parse(msg.payload).get<Protocol::AutoAddMsg_t>();

    if (autoAdd.dev.type == "autoaddstart") {
        _device->deviceOpenBleBroadcast(autoAdd.dev.param);
        _blueTooth->leScanStart();
    } else if (autoAdd.dev.type == "autoaddstop") {
        _blueTooth->leScanStop();
        _device->clearAutoFindTemp();
    } else if (autoAdd.dev.type == "autoaddrestart") {
        _device->deviceOpenBleBroadcast(autoAdd.dev.param);
        // _blueTooth->leScanStop();
        _device->clearAutoFindTemp();
        _blueTooth->leScanStart();
    }

    return true;
}

bool MainProcess::devOtaMsgParse(const Msg_t &msg)
{
    auto result = split(msg.topic, "/");
    auto &dest = result[2];
    auto &src = result[1];

    Protocol::Upgrade::DeviceUpgradeAck_t ack = json::parse(msg.payload).get<Protocol::Upgrade::DeviceUpgrade_t>();

    ack.setSrc(dest);

    if (auto dev = _device->findModbusDeviceById(ack.getId()); dev) {
        if (dev->upgrade(ack.getPath())) {
            ack.setStatus(true);
        } else {
            ack.setStatus(false);
        }

        std::string ackTopic = "dev/" + dest + "/" + src + "/otaack";

        return _mqtt->publish(ackTopic, json(ack).dump());
    }

    main_error("can't find device, device id --> {}", dest);

    return true;
}

bool MainProcess::devDeleteById(const Msg_t &msg)
{
    auto result = split(msg.topic, "/");
    auto &dest = result[2];
    auto &src = result[1];

    if (dest != devModbusSrc) {
        return true;
    }

    Protocol::DeleteMsg_t del = json::parse(msg.payload).get<Protocol::DeleteMsg_t>();

    Protocol::DeleteMsgAck_t ack = del;
    ack.setSrc(devModbusSrc);

    for (auto &dev : ack._dev.devices) {
        _device->delModbusDevice(dev.id);
        dev.setStatus("success");
    }

    std::string ackTopic = "dev/" + dest + "/" + src + "/deleteack";

    return _mqtt->publish(ackTopic, json(ack).dump());
}

void MainProcess::waitMqttMsgThread(void)
{
    while (true) {
        auto msg = _mqtt->getOneItem();
        _pool->exec(&MainProcess::mqttMsgParse, this, msg);
    }
}

void MainProcess::requestModbusRlistThread(void)
{
    do {
        if (!_isRecvModbusRlistAck) {
            _mqtt->publish(devModbusRlistTopic, json(Protocol::RlistMsg_t(devModbusSrc, devModbusType, 0, 32)).dump());
        } else {
            break;
        }
        std::this_thread::sleep_for(std::chrono::seconds(5));
    } while (true);
}

void MainProcess::request645RlistThread(void)
{
    do {
        if (!_isRecvDl645RlistAck) {
            _mqtt->publish(devModbusRlistTopic, json(Protocol::RlistMsg_t(devModbusSrc, dev645Type, 0, 32)).dump());
        } else {
            break;
        }
        std::this_thread::sleep_for(std::chrono::seconds(5));
    } while (true);
}

void MainProcess::releaseDeviceListThread(void)
{
    while (true) {
        std::unique_lock<std::mutex> _lock(_releaseMutex);
        _releaseCv.wait(_lock, [&] { return !_releaseDeviceList.empty(); });

        for (const auto &id : _releaseDeviceList) {
            auto dev = _device->findModbusDeviceById(id);
            if (!dev) {
                continue;
            }

            auto point = dev->pointTable.findPoint("Mac");
            if (point.has_value()) {
                if (dev->readPoint("Mac", true)) {
                    dev->setMac(dev->pointTable["Mac"].rData.toMacString());
                    // std::string devBroadcast0InfoTopic = "dev/" + std::to_string(dev->id.value()) + "/0/info";
                    // _mqtt->publish(devBroadcast0InfoTopic, json(Protocol::Broadcast0Info_t(devModbusSrc, dev->id.value(), true).setKeyValue("Mac", dev->mac)).dump());
                }
            } else {
                auto p = dev->pointTable.findFirstReadAblePoint();

                if (p.has_value()) {
                    dev->readPoint(p.value()->first, true);
                    // std::string devBroadcast0InfoTopic = "dev/" + std::to_string(dev->id.value()) + "/0/info";
                    // _mqtt->publish(devBroadcast0InfoTopic, json(Protocol::Broadcast0Info_t(devModbusSrc, dev->id.value(), true).setKeyValue(p.value()->first, p.value()->second.rData.get<uint32_t>())).dump());
                }
            }
        }

        std::this_thread::sleep_for(std::chrono::milliseconds(500));
    }
}

std::vector<std::string> MainProcess::split(const std::string &source, const std::string &s)
{
    std::vector<std::string> result;

    std::string::size_type pos = 0;
    std::string::size_type prev = 0;

    while ((pos = source.find(s, prev)) != std::string::npos) {
        auto r = source.substr(prev, pos - prev);

        r.size() ? result.push_back(r) : (void)(nullptr);

        prev = pos + s.length();
    }
    auto r = source.substr(prev);
    r.size() ? result.push_back(r) : (void)(nullptr);

    return result;
}

void MainProcess::publish0Info(void *dev, const std::string &name, OldData_t &oldData, NewData_t &newData)
{
    auto device = static_cast<ModbusDevice_t *>(dev);

    auto point = device->pointTable.findPoint(name);
    if (!point.has_value()) {
        return;
    }

#if defined(READ_POINT_CHANGE_REPORT)
    if (oldData == newData) {
        return;
    }
#endif

    Protocol::Broadcast0Info_t info(devModbusSrc, device->id.value(), device->isOnline());

    if (point.value()->type == TYPE_U32 || point.value()->type == TYPE_U64 || point.value()->type == TYPE_U16 || point.value()->type == TYPE_U8 || point.value()->type == TYPE_BITS) {
        if (point.value()->unit.has_value()) {
            info.setKeyValue(name, newData.get<uint32_t>() * point.value()->unit.value());
        } else {
            info.setKeyValue(name, newData.get<uint32_t>());
        }
    } else if (point.value()->type == TYPE_I64 || point.value()->type == TYPE_I32 || point.value()->type == TYPE_I16 || point.value()->type == TYPE_I8) {
        if (point.value()->unit.has_value()) {
            info.setKeyValue(name, newData.get<int>() * point.value()->unit.value());
        } else {
            info.setKeyValue(name, newData.get<int>());
        }
    } else if (point.value()->type == TYPE_STR) {
        info.setKeyValue(name, newData.get<std::string>());
    } else if (point.value()->type == TYPE_STR_HEX) {
        info.setKeyValue(name, newData.toHexString());
    }

    std::string devBroadcast0InfoTopic = "dev/" + std::to_string(device->id.value()) + "/0/info";

    _mqtt->publish(devBroadcast0InfoTopic, json(info).dump());
}

void MainProcess::onLineStatChange(void *arg)
{
    auto device = static_cast<ModbusDevice_t *>(arg);

    std::string devBroadcast0InfoTopic = "dev/" + std::to_string(device->id.value()) + "/0/info";

    if (device->isOnline()) { /* offline --> online */
        std::cout << "device online !!!!" << std::endl;
        // auto point = device->pointTable.findFirstReadAblePoint();
        // if (point.has_value()) {
        //     auto name = point.value()->first;
        //     _mqtt->publish(devBroadcast0InfoTopic, json(Protocol::Broadcast0Info_t(devModbusSrc, device->id.value(), false).setKeyValue(name, 0)).dump());
        // }

        _releaseDeviceList.erase(std::remove_if(_releaseDeviceList.begin(), _releaseDeviceList.end(), [device](std::size_t id) { return id == device->id.value(); }), _releaseDeviceList.end());

    } else {
        std::cout << "device offline !!!!" << std::endl;
        auto point = device->pointTable.findFirstReadAblePoint();
        if (point.has_value()) {
            auto name = point.value()->first;
            _mqtt->publish(devBroadcast0InfoTopic, json(Protocol::Broadcast0Info_t(devModbusSrc, device->id.value(), false).setKeyValue(name, 0)).dump());
        }

        if (std::find_if(_releaseDeviceList.begin(), _releaseDeviceList.end(), [device](std::size_t id) { return id == device->id.value(); }) == _releaseDeviceList.end()) {
            _releaseDeviceList.push_back(device->id.value());
            _releaseCv.notify_one();
        }
    }
}

void MainProcess::onAddModbusDevice(std::shared_ptr<ModbusDevice_t> device)
{
    /* bind 0 info callback */
    for (auto &point : device->pointTable) {
        addPointOperationCallback(point.second, std::bind(&MainProcess::publish0Info, this, std::placeholders::_1, std::placeholders::_2, std::placeholders::_3, std::placeholders::_4));
    }
    device->registerOnlineStateChangeCallback(std::bind(&MainProcess::onLineStatChange, this, std::placeholders::_1));

    _releaseDeviceList.push_back(device->id.value());
    _releaseCv.notify_one();
}

void MainProcess::onDelModbusDevice(std::shared_ptr<ModbusDevice_t> device)
{
    _releaseDeviceList.erase(std::remove_if(_releaseDeviceList.begin(), _releaseDeviceList.end(), [device](std::size_t id) { return id == device->id.value(); }), _releaseDeviceList.end());
}

std::vector<uint8_t> MainProcess::mac2vec(const std::string &mac)
{
    auto result = split(mac, ":");
    std::vector<uint8_t> v;
    for (const auto &item : result) {
        v.emplace_back(std::stoi(item, 0, 16));
    }

    return v;
}

void MainProcess::onLeScanDevice(const std::string &name, const std::string &mac)
{
    if (name.find("NB2") == std::string::npos) {
        return;
    }

    main_info("find name--> {} && find mac --> {}", name, mac);

    if (_device->findModbusDeviceByMac(mac)) {
        main_info("device already exist");
        return;
    }

    auto dev = _device->autoFindDevice(mac2vec(mac));
    if (dev.has_value()) {
        dev->setMac(mac);
        _device->addDeviceToAutoFindTemp(dev.value());

        auto model = dev->pointTable.findPoint("prodModel").value()->rData.toHexString();
        auto sn = dev->pointTable.findPoint("sn").value()->rData.get<std::string>();
        if (_device->findModbusDeviceBySn(sn)) {
            main_info("device already exist");
            return;
        }

        Protocol::AutoFindAck_t ack(devModbusSrc, sn, model, dev.value().hardware, dev.value().slaveAddr);

        _mqtt->publish(devAutoAddAckTopic, json(ack).dump());
    }
}
